package com.hzgj.bcl.soa.cicada.router;

import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.google.common.collect.Sets;

import com.hzgj.bcl.soa.cicada.Lifecycle;
import com.hzgj.bcl.soa.cicada.pool.ChannelPoolConfig;
import com.hzgj.bcl.soa.cicada.pool.ChannelPoolFactory;
import com.hzgj.bcl.soa.cicada.pool.ChannelPoolManager;
import com.hzgj.bcl.soa.cicada.pool.StatisticChannelPool;
import com.hzgj.bcl.soa.cicada.sender.CicadaSender;
import com.hzgj.bcl.soa.cicada.sender.Sender;
import com.hzgj.bcl.soa.config.ChannelConfig;
import com.hzgj.bcl.soa.config.MethodDef;
import com.hzgj.bcl.soa.config.ReferenceDef;
import com.hzgj.bcl.soa.config.ServerDef;
import com.hzgj.bcl.soa.config.ServiceDef;
import com.hzgj.bcl.soa.registry.NotifyListener;
import com.hzgj.bcl.soa.registry.RegistryService;
import com.hzgj.bcl.soa.rpc.IpAddress;
import com.hzgj.bcl.soa.rpc.ServiceAddress;
import com.hzgj.bcl.soa.util.Pair;
import com.hzgj.bcl.util.MapUtil;
import com.hzgj.bcl.util.jmx.annotation.MBean;
import com.hzgj.bcl.util.jmx.annotation.ManagedOperation;
import com.hzgj.bcl.util.lang.StrKit;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.Serializable;
import java.net.UnknownHostException;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.Callable;
import java.util.concurrent.CancellationException;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.FutureTask;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.ReentrantLock;
import java.util.stream.Collectors;

import io.netty.channel.pool.ChannelPool;
import io.netty.util.HashedWheelTimer;
import lombok.Builder;
import lombok.Getter;

import static com.hzgj.bcl.soa.cicada.pool.ChannelPoolManager.DEFAULT_POOL_KEY;

/**
 * Created by warning5 on 2017/3/3.
 */
@MBean(description = "Cicada Router")
public class CicadaRouter implements NotifyListener, MutableRouter {

    ChannelPoolFactory channelPoolFactory = ChannelPoolFactory.getInstance();
    private Logger logger = LoggerFactory.getLogger(getClass());
    private ReentrantLock lock = new ReentrantLock();
    private ConcurrentMap<ReferenceDef, Set<Sender>> upServers = Maps.newConcurrentMap();
    private ConcurrentMap<ReferenceDef, Set<Sender>> markedDownServers = Maps.newConcurrentMap();
    private Map<IpAddress, CicadaSender> ipSenders = Maps.newHashMap();
    private Map<ReferenceDef, ServiceAddress> serviceAddresses;
    private Map<String, ReferenceDef> referenceDefs;
    private HashedWheelTimer timer = new HashedWheelTimer();
    private Map<String, ChannelConfig> channelConfigs;
    private Map<String, ChannelPoolConfig> channelPoolConfigs;
    private RegistryService registryService;
    private ConcurrentMap<ReferenceDef, Future<Set<Sender>>> unCheckServices;

    public CicadaRouter(final Map<ReferenceDef, ServiceAddress> serviceAddresses, Map<String, ChannelPoolConfig> channelPoolConfigs,
                        Map<String, ChannelConfig> channelConfigs, RegistryService registryService) {
        this.serviceAddresses = serviceAddresses;
        this.channelConfigs = channelConfigs;
        this.channelPoolConfigs = channelPoolConfigs;
        this.referenceDefs = serviceAddresses.keySet().stream().collect(Collectors.toMap(ReferenceDef::getService, referenceDef -> referenceDef));
        this.registryService = registryService;
    }

    @Override
    public void start() {
        final AtomicBoolean unCheck = new AtomicBoolean(false);
        serviceAddresses.entrySet().stream().filter(entry -> {
            boolean check = entry.getKey().isCheck();
            if (!check) {
                unCheck.set(true);
            }
            return check;
        }).forEach(entry -> {
            ReferenceDef referenceDef = entry.getKey();
            ChannelConfig channelConfig = getChannelConfig(referenceDef.getProtocol(), referenceDef.getService());
            entry.getValue().list().forEach(ipAddress -> {
                CicadaSender sender = ipSenders.get(ipAddress);
                if (sender == null) {
                    sender = new CicadaSender(timer, ipAddress);
                    ChannelPoolManager channelPoolManager = new ChannelPoolManager();
                    sender.channelPoolManager(channelPoolManager).start();
                    ipSenders.put(ipAddress, sender);
                }
                initSender(referenceDef, channelConfig, ipAddress, sender);
                putMapValue(upServers, referenceDef, sender);
            });
        });
        if (registryService != null) {
            registryService.subscribeServices(this,
                    referenceDefs.values().stream().filter(referenceDef -> StrKit.isBlank(referenceDef.getUrl()) && referenceDef.isCheck())
                            .collect(Collectors.toList()).toArray(new ReferenceDef[0]));
        }

        if (unCheck.get()) {
            unCheckServices = Maps.newConcurrentMap();
        }
    }


    @Override
    public void stop() {
        upServers.values().forEach(senders -> senders.forEach(Lifecycle::stop));
        markedDownServers.values().forEach(senders -> senders.forEach(Lifecycle::stop));
        if (registryService != null) {
            registryService.unSubscribeServices(referenceDefs.values().stream().filter(referenceDef -> StrKit.isBlank(referenceDef.getUrl()))
                    .collect(Collectors.toList()).toArray(new ReferenceDef[0]));
        }
        timer.stop();
        unCheckServices = null;
        upServers = null;
        markedDownServers = null;
        ipSenders = null;
        serviceAddresses = null;
        referenceDefs = null;
        channelConfigs = null;
        channelPoolConfigs = null;
    }

    @Override
    public Set<Sender> getUpSenders(ReferenceDef param) {
        ReferenceDef referenceDef = referenceDefs.get(param.getService());
        Set<Sender> senders = upServers.get(referenceDef);
        if (senders == null && !referenceDef.isCheck()) {
            if (StrKit.isBlank(referenceDef.getUrl())) {
                if (registryService != null) {
                    registryService.subscribeServices(this, referenceDef);
                }
            }
            try {
                senders = getReferenceSender(referenceDef);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                return Collections.emptySet();
            }
        } else {
            if (!referenceDef.isCheck()) {
                //删除已初始化的sender
                unCheckServices.remove(referenceDef);
            }
        }
        return Collections.unmodifiableSet(senders);
    }

    @Override
    public Set<Sender> getMarkedDownSenders(ReferenceDef referenceDef) {
        Set<Sender> senders = markedDownServers.get(referenceDef);
        if (senders != null) {
            return Collections.unmodifiableSet(senders);
        } else {
            return Collections.emptySet();
        }
    }

    @Override
    public void markDown(String service, Sender sender) {
        ReferenceDef referenceDef = referenceDefs.get(service);
        if (referenceDef != null) {
            upServers.get(referenceDef).remove(sender);
            sender.setStatus(Sender.Status.SUSPEND);
            putMapValue(markedDownServers, referenceDef, sender);
        } else {
            logger.error("can't find service def of " + service + " in router.");
        }
    }

    @ManagedOperation(description = "get pool info")
    public List<ClientPoolInfo> getPoolInfo() {
        List<ClientPoolInfo> result = Lists.newArrayList();
        ipSenders.forEach((ipAddress, cicadaSender) -> {
            ClientPoolInfo.ClientPoolInfoBuilder builder = ClientPoolInfo.builder().address(ipAddress.toString());
            cicadaSender.getChannelPoolManager().getPools().forEach(channelPool -> {
                StatisticChannelPool statisticChannelPool = (StatisticChannelPool) channelPool;
                result.add(builder.max(statisticChannelPool.getMaxSize()).current(statisticChannelPool.getWorkingSize()).build());
            });
        });
        return result;
    }

    @Getter
    @Builder
    public static class ClientPoolInfo implements Serializable {
        String address;
        int max;
        int current;
        String key;
    }

    @ManagedOperation(description = "update reference config in router,only support version and group")
    public Pair<Boolean, String> updateReference(List<ReferenceDef> refs) {
        StringBuilder builder = new StringBuilder();
        AtomicBoolean status = new AtomicBoolean(true);
        refs.forEach(ref -> {
            ReferenceDef referenceDef = referenceDefs.get(ref.getService());
            if (referenceDef != null) {
                try {
                    if (lock.tryLock(2, TimeUnit.SECONDS)) {
                        logger.info("change client " + referenceDef.getHost() + " for service " + referenceDef.getService()
                                + " version from " + referenceDef.getVersion() + " to " + ref.getVersion()
                                + " and group from " + referenceDef.getGroup() + " to " + ref.getGroup());
                        referenceDef.setGroup(ref.getGroup());
                        referenceDef.setVersion(ref.getVersion());
                    } else {
                        status.set(false);
                        builder.append("can't get update lock of " + ref.getService() + ",try it later.").append("\n");
                    }
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                } finally {
                    lock.unlock();
                }
            } else {
                status.set(false);
                builder.append("can't find referenceDef of ").append(ref.getService()).append("\n");
            }
        });
        registryService.getRegistry().updateClient(referenceDefs.values());
        return new Pair<>(status.get(), builder.toString());
    }

    private ChannelPool getPool(ChannelPoolFactory channelPoolFactory, ChannelConfig
            channelConfig, IpAddress ipAddress, String poolId, String service, String poolKey, CicadaSender sender) {
        ChannelPoolConfig channelPoolConfig = getChannelPoolConfig(poolId, service);
        ChannelPoolFactory.ChannelPoolBuilder channelPoolBuilder = ChannelPoolFactory.ChannelPoolBuilder.builder().
                sender(sender).channelConfig(channelConfig).channelPoolConfig(channelPoolConfig)
                .ipAddress(ipAddress).poolKey(poolKey).build();
        return channelPoolFactory.getChannelPool(channelPoolBuilder);
    }

    private ChannelConfig getChannelConfig(String protocol, String service) {
        ChannelConfig channelConfig;
        if (StrKit.notBlank(protocol)) {
            channelConfig = channelConfigs.get(protocol);
            if (channelConfig == null) {
                logger.error("reference interface " + service + " protocol is invalidate,will use default channel config.");
                channelConfig = new ChannelConfig();//Default channelConfig
            }
        } else {
            channelConfig = new ChannelConfig();
        }
        return channelConfig;
    }

    private ChannelPoolConfig getChannelPoolConfig(String pool_id, String service) {
        ChannelPoolConfig channelPoolConfig;
        if (StrKit.notBlank(pool_id)) {
            channelPoolConfig = channelPoolConfigs.get(pool_id);
            if (channelPoolConfig == null) {
                logger.error("reference interface " + service + " pool is invalidate,will use default channelPool config.");
                channelPoolConfig = new ChannelPoolConfig();//Default channelPoolConfig
            }
        } else {
            channelPoolConfig = new ChannelPoolConfig();
        }
        return channelPoolConfig;
    }

    private Set<Sender> getReferenceSender(ReferenceDef referenceDef) throws InterruptedException {
        while (true) {
            Future<Set<Sender>> f = unCheckServices.get(referenceDef);
            if (f == null) {
                Callable<Set<Sender>> eval = () -> {
                    Set<Sender> senders = Sets.newHashSet();
                    ServiceAddress serviceAddress = ServiceAddress.getAddress(referenceDef, registryService);
                    for (IpAddress ipAddress : serviceAddress.list()) {
                        if (ipSenders.containsKey(ipAddress)) {
                            senders.add(ipSenders.get(ipAddress));
                        } else {
                            Sender sender = newSender(referenceDef, ipAddress);
                            putMapValue(upServers, referenceDef, sender);
                            senders.add(sender);
                        }
                    }
                    return senders;
                };
                FutureTask<Set<Sender>> ft = new FutureTask<>(eval);
                f = unCheckServices.putIfAbsent(referenceDef, ft);
                if (f == null) {
                    f = ft;
                    ft.run();
                }
            }
            try {
                return f.get();
            } catch (CancellationException e) {
                unCheckServices.remove(referenceDef, f);
            } catch (ExecutionException e) {
                logger.error("{}", e);
                throw new RuntimeException(e);
            }
        }
    }

    @Override
    public void notifyWithSingleService(List<ServerDef> servers) {
        Map<ReferenceDef, Set<Sender>> enters = Maps.newHashMap();
        servers.stream().forEach(server -> {
            try {
                IpAddress ipAddress = new IpAddress(server.getHost(), server.getPort());
                server.getServiceDefs().forEach(attr -> {
                    ReferenceDef referenceDef = referenceDefs.get(attr.getInterface());
                    if (enableGroupAndGroup(attr, referenceDef, 0)) {
                        buildConsistentWithVersionAndGroup(ipAddress, referenceDef);
                        putMapValue(enters, referenceDef, ipSenders.get(ipAddress));//for new add ipAddress
                    } else {
                        removeSenders(attr, ipAddress, referenceDef);
                    }
                });
            } catch (Exception e) {
                logger.error("{}", e);
            }
        });

        if (enters.size() == 0) {
            return;
        }
        diffHandle(upServers, enters, "remove service provider ");
        if (markedDownServers.size() == 0) {
            return;
        }
        diffHandle(markedDownServers, enters, "remove markedDown service provider ");
    }

    @Override
    public void notifyClient(ServiceDef.Attr attr, PushAction action) {
        try {
            ReferenceDef referenceDef = referenceDefs.get(attr.getInterface());
            IpAddress ipAddress = new IpAddress(attr.getServerDef().getHost(), attr.getServerDef().getPort());
            if (action.equals(PushAction.set)) {
                if (enableGroupAndGroup(attr, referenceDef, 0)) {
                    buildConsistentWithVersionAndGroup(ipAddress, referenceDef);
                } else {
                    removeSenders(attr, ipAddress, referenceDef);
                }
            } else if (action.equals(PushAction.expire)) {
                removeSenders(attr, ipAddress, referenceDef);
            } else if (action.equals(PushAction.delete)) {
                removeSenders(attr, ipAddress, referenceDef);
            }
        } catch (UnknownHostException e) {
            logger.error("{}", e);
        }
    }

    private void buildConsistentWithVersionAndGroup(IpAddress ipAddress, ReferenceDef referenceDef) {
        Sender sender = ipSenders.get(ipAddress);
        if (sender != null) {
            if (!contain(upServers.get(referenceDef), sender)) {
                if (contain(markedDownServers.get(referenceDef), sender)) {
                    markedDownServers.get(referenceDef).remove(sender);
                    sender.setStatus(Sender.Status.OK);
                    upServers.get(referenceDef).add(sender);
                } else {
                    putMapValue(upServers, referenceDef, sender);
                    logger.info("add service provider " + sender + " for " + referenceDef.getService());
                }
            }
        } else {
            //notifyWithSingleService在并发情况下：
            //         A thread                                       B thread
            // 1、newSender(referenceDef, ipAddress)
            //                                            1、 removeSender方法判断upServers或
            //                                               markedDownServers发现没有sender引用该ipAddress
            // 2、putMapValue(upServers)
            //   （添加ipAddress到upServers导致ipSenders中
            //   不存在sender，但upServers中存在）
            Sender newSender = newSender(referenceDef, ipAddress);
            putMapValue(upServers, referenceDef, newSender);
            logger.info("add service provider " + newSender + " for " + referenceDef.getService());
        }
    }

    private void removeSenders(ServiceDef.Attr attr, IpAddress ipAddress, ReferenceDef referenceDef) {
        Set<Sender> senders = upServers.get(referenceDef);
        if (senders != null && senders.size() > 0) {
            if (senders.size() > 1) {
                removeSenderByAddress(upServers, referenceDef, ipAddress,
                        String.format("remove invalidate service provider,client group=%s,version=%s,remote group=%s,version=%s",
                                referenceDef.getGroup(), referenceDef.getVersion(), attr.getGroup(), attr.getVersion()));
            } else {
                if (senders.size() == 1) {
                    Sender sender = Lists.newArrayList(senders).get(0);
                    if (sender.getIpAddress().equals(ipAddress)) {
                        logger.error("can't remove only one provider " + ipAddress + " for service " + referenceDef.getService()
                                + " remote provider(version=" + attr.getVersion() + ",group=" + attr.getGroup() + ")," +
                                " local provider(version=" + referenceDef.getVersion() + ",group=" + referenceDef.getGroup() + ")");

                    }
                }
            }
        }

        if (markedDownServers.containsKey(referenceDef)) {
            removeSenderByAddress(markedDownServers, referenceDef, ipAddress,
                    String.format("remove invalidate markedDown service provider,client group=%s,version=%s,remote group=%s,version=%s",
                            referenceDef.getGroup(), referenceDef.getVersion(), attr.getGroup(), attr.getVersion()));
        }
    }

    private void diffHandle(Map<ReferenceDef, Set<Sender>> servers, Map<ReferenceDef, Set<Sender>> enters, String message) {
        enters.entrySet().stream()
                .filter(enter -> servers.containsKey(enter.getKey()))
                .forEach(entry -> {
                    Set<Sender> existSenders = servers.get(entry.getKey());
                    Sets.SetView<Sender> diff = Sets.difference(existSenders, entry.getValue());
                    if (diff.size() > 0) {
                        Set<Sender> _removed = Sets.newHashSet();
                        diff.copyInto(_removed);
                        existSenders.removeAll(_removed);
                        _removed.stream().forEach(sender -> {
                            if (!sender.isClose()) {
                                removeSender(servers, entry.getKey(), sender, message);
                            } else {
                                logger.info(message + "for " + sender.getIpAddress() + " of " + entry.getKey().getService());
                            }
                        });
                    }
                });
    }

    private void removeSenderByAddress(Map<ReferenceDef, Set<Sender>> servers, ReferenceDef referenceDef, IpAddress ipAddress, String message) {
        Sender sender = ipSenders.get(ipAddress);
        if (sender == null) {
            return;
        }
        Set<Sender> senders = servers.get(referenceDef);
        senders.remove(sender);
        removeSender(servers, referenceDef, sender, message);
    }


    boolean enableGroupAndGroup(ServiceDef.Attr attr, ReferenceDef referenceDef, int retry) {
        try {
            if (lock.tryLock(2, TimeUnit.SECONDS)) {
                return enableGroupAndGroup0(attr, referenceDef);
            } else {
                if (retry < 2) {
                    return enableGroupAndGroup(attr, referenceDef, ++retry);
                }
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        } finally {
            lock.unlock();
        }
        logger.info("can't get clientConfig lock,will retry in next span.");
        return false;
    }

    boolean enableGroupAndGroup0(ServiceDef.Attr attr, ReferenceDef referenceDef) {
        if (StrKit.isBlank(referenceDef.getVersion()) && StrKit.isBlank(referenceDef.getGroup())) {
            if (StrKit.notBlank(attr.getGroup()) || StrKit.notBlank(attr.getVersion())) {
                return false;
            } else {
                return true;
            }
        } else if (StrKit.isBlank(referenceDef.getVersion()) && StrKit.notBlank(referenceDef.getGroup())) {
            return referenceDef.getGroup().equals(attr.getGroup());
        } else if (StrKit.notBlank(referenceDef.getVersion()) && StrKit.isBlank(referenceDef.getGroup())) {
            return referenceDef.getVersion().equals(attr.getVersion());
        } else {
            return referenceDef.getVersion().equals(attr.getVersion()) && referenceDef.getGroup().equals(attr.getGroup());
        }
    }

    private boolean contain(Set<Sender> servers, Sender sender) {
        if (servers == null) {
            return false;
        }
        return servers.contains(sender);
    }

    /**
     * 该方法无法删除sender，避免抛出ConcurrentModificationException
     */
    private void removeSender(Map<ReferenceDef, Set<Sender>> servers, ReferenceDef referenceDef, Sender sender, String message) {
        IpAddress ipAddress = sender.getIpAddress();
        Set<Sender> senders = servers.get(referenceDef);
        if (senders.size() == 0) {
            servers.remove(referenceDef);
        }
        //多个服务可能都依赖某个sender
        long count = servers.values().stream().filter(_senders -> _senders.contains(sender)).count();
        if (count == 0) {
            ipSenders.remove(ipAddress);
            channelPoolFactory.removeChannelPool(ChannelPoolFactory.ChannelPoolBuilder.builder().ipAddress(ipAddress).poolKey(DEFAULT_POOL_KEY).build());
            if (referenceDef.getMethodDefs() != null) {
                referenceDef.getMethodDefs().values().stream().filter(methodDef -> StrKit.notBlank(methodDef.getPool())).forEach(methodDef -> {
                    channelPoolFactory.removeChannelPool(ChannelPoolFactory.ChannelPoolBuilder.builder().ipAddress(ipAddress).poolKey(referenceDef.getService() + "_" + methodDef.getName()).build());
                });
            }
            sender.stop();
        } else {
            if (referenceDef.getMethodDefs() != null) {
                referenceDef.getMethodDefs().values().stream().filter(methodDef -> StrKit.notBlank(methodDef.getPool())).forEach(methodDef -> {
                    channelPoolFactory.removeChannelPool(ChannelPoolFactory.ChannelPoolBuilder.builder().ipAddress(ipAddress).poolKey(referenceDef.getService() + "_" + methodDef.getName()).build());
                });
            }
        }
        logger.info(message + "for " + ipAddress + " of " + referenceDef.getService());
    }

    private Sender newSender(ReferenceDef referenceDef, IpAddress ipAddress) {
        CicadaSender sender = new CicadaSender(timer, ipAddress);
        ChannelPoolManager channelPoolManager = new ChannelPoolManager();
        sender.channelPoolManager(channelPoolManager).start();
        ipSenders.put(ipAddress, sender);
        initSender(referenceDef, getChannelConfig(referenceDef.getProtocol(), referenceDef.getService()), ipAddress, sender);
        return sender;
    }

    private void initSender(ReferenceDef referenceDef, ChannelConfig channelConfig, IpAddress ipAddress, CicadaSender sender) {
        if (referenceDef.getMethodDefs() != null) {
            int count = 0;
            for (MethodDef methodDef : referenceDef.getMethodDefs().values()) {
                if (StrKit.notBlank(methodDef.getPool())) {
                    count++;
                    initSender(referenceDef, channelConfig, methodDef.getPool(), ipAddress, methodDef.getName(), sender);
                }
            }
            if (referenceDef.getMethodCount() > count) {
                initSender(referenceDef, channelConfig, referenceDef.getPool(), ipAddress, "", sender);
            }
        } else {
            initSender(referenceDef, channelConfig, referenceDef.getPool(), ipAddress, "", sender);
        }
    }

    private void initSender(ReferenceDef referenceDef, ChannelConfig channelConfig, String poolId, IpAddress ipAddress, String method, CicadaSender _sender) {
        String key = DEFAULT_POOL_KEY;
        if (StrKit.notBlank(method)) {
            key = referenceDef.getService() + "_" + method;
        }
        ChannelPool channelPool = getPool(channelPoolFactory, channelConfig, ipAddress, poolId, referenceDef.getService(), key, _sender);
        _sender.getChannelPoolManager().addChannelPool(key, channelPool);
    }

    private <T> void putMapValue(Map<ReferenceDef, Set<T>> servers, ReferenceDef referenceDef, T value) {
        MapUtil.putMapValue(servers, referenceDef, value);
    }
}
